package extnode

import (
	"errors"
	"os"
	"strconv"
	"time"
	"util/zkwrapper"
)

//grasp follower addresses from zk
func (s *Server) detectFollowersAllOnboarding() bool {
	for j := 1; j <= FullReplFactor; j++ {
		addr, e := s.zkw.GetServerValue("/jfs-root/extnode-rg/"+strconv.Itoa(s.rg), s.id+j)
		if e != nil {
			s.errLog.Printf("Error: follower %v is missing\n", s.id+j)
			return false
		}

		if s.fcp.followerAddr[j-1] != addr {
			s.fcp.followerAddr[j-1] = addr
		}
	}

	return true
}

//check disk exceptions - full or corrupt
func (s *Server) checkDiskStatus() bool {
	if s.store.StatFreeSpace() == 0 {
		s.errLog.Println("Fatal: the disk is to be full")
		s.setState(DiskAlreadyFull)
		s.setAvailabilityLevel(DiskAlreadyFull)
		return false
	}

	if s.stats.DiskErrors() > MaxDiskErrors {
		s.errLog.Println("Fatal: too many disk errors so switch to unavailable")
		s.setState(DiskCorrupted)
		s.setAvailabilityLevel(DiskCorrupted)
		return false
	}
	return true
}

//connect Zk and add the paths
func (s *Server) registerWithZk() error {
	var e error
	s.zkw, e = zkwrapper.NewZkWrapper(s.zk)
	if e != nil {
		s.zkw = nil
		s.errLog.Println("Fatal: failed to connect ZK ", e)
		return e
	}

	//wait for zk session expiry
	<-time.After(NormalWakeUpPeriod * time.Second)

	//add itself to its own repl dir
	e = s.zkw.AddEphemeralServer("/jfs-root/extnode-rg/"+strconv.Itoa(s.rg), s.id, s.ip+":"+s.port)
	if e != nil {
		s.zkw.Close()
		s.zkw = nil
		s.errLog.Println("Fatal: failed to register with ZK ", e)
		return e
	}

	//add to client-visible dir
	weight := WeightFactor(s.store.StatFreeSpace(), s.getState())
	e = s.zkw.AddEphemeralServer("/jfs-root/extnode", s.id, s.ip+":"+s.port+","+strconv.FormatInt(weight, 10)+"|"+s.dc)
	if e != nil {
		s.zkw.Close()
		s.zkw = nil
		s.errLog.Println("Fatal: failed to register with ZK ", e)
		return e
	}

	s.infoLog.Println("Info: register with Zk")
	return nil
}

//return false if zk conn was broken
func (s *Server) checkZkConnHealth() bool {
	if s.zkw == nil {
		return s.registerWithZk() == nil
	}

	if s.zkw.CheckExistingNode("/jfs-root/extnode/"+strconv.Itoa(s.id)) &&
		s.zkw.CheckExistingNode("/jfs-root/extnode-rg/"+strconv.Itoa(s.rg)+"/"+strconv.Itoa(s.id)) {
		return true
	}

	s.infoLog.Println("Info: Zk connection was broken so re-connect")
	s.zkw.Close()
	return s.registerWithZk() == nil
}

func (s *Server) setAvailabilityLevel(state int32) error {
	if s.zkw == nil {
		s.errLog.Println("Error: zk is not available")
		return errors.New("ZkUnavailable")
	}

	weight := WeightFactor(s.store.StatFreeSpace(), s.getState())
	err := s.zkw.SetServerValue("/jfs-root/extnode", s.id, s.ip+":"+s.port+","+strconv.FormatInt(weight, 10)+"|"+s.dc)
	if err == nil {
		s.infoLog.Printf("Info: update to Zk with weight %v\n", weight)
		return nil
	}

	//close the handle in case of errors
	s.zkw.Close()
	s.zkw = nil
	s.errLog.Println("Fatal: failed to update Zk")
	return err
}

func (s *Server) checkLogRotation() {
	var fp *os.File
	const LogFileOpt = os.O_RDWR | os.O_CREATE | os.O_APPEND

	_, err := os.Stat(s.metadir + "/" + ReadLogFileName + "." + time.Now().Format("2006-01-02"))
	if err == nil {
		return
	}

	//rotate the three log files
	fp, err = os.OpenFile(s.metadir+"/"+ReadLogFileName+"."+time.Now().Format("2006-01-02"), LogFileOpt, 0666)
	if err == nil {
		s.readLog.SetOutput(fp)
	}

	fp, err = os.OpenFile(s.metadir+"/"+WriteLogFileName+"."+time.Now().Format("2006-01-02"), LogFileOpt, 0666)
	if err == nil {
		s.writeLog.SetOutput(fp)
	}

	fp, err = os.OpenFile(s.metadir+"/"+WorkLogFileName+"."+time.Now().Format("2006-01-02"), LogFileOpt, 0666)
	if err == nil {
		s.workLog.SetOutput(fp)
	}
}

func (s *Server) serveReplGroupChange() {
	for {
		//report stats to the monitor
		s.ReportStats()

		//check log rotation
		s.checkLogRotation()

		//check disk corruption and zk exception
		if s.checkDiskStatus() == false || s.checkZkConnHealth() == false {
			<-time.After(AbnormalWakeUpPeriod * time.Minute)
			continue
		}

		noFollowerAbsent := s.detectFollowersAllOnboarding()

		switch {
		case noFollowerAbsent && s.getState() != ReplGroupReady:
			//make the replica group ready
			s.setState(ReplGroupReady)
			s.setAvailabilityLevel(ReplGroupReady)
			s.fcp.flushAll()

		case noFollowerAbsent == false && s.getState() == ReplGroupReady:
			s.setState(ReplGroupSplit)
			s.setAvailabilityLevel(ReplGroupSplit)
			s.fcp.flushAll()

		default:
			//keep the current state
			s.setAvailabilityLevel(s.getState())
		}

		//in case of zk errors, zkw was cleared
		if s.zkw == nil {
			s.errLog.Println("Fatal: zkw was cleared so continue")
			<-time.After(AbnormalWakeUpPeriod * time.Minute)
			continue
		}

		//some follower is missing, so watch the repl-group dir on Zk
		ch, e := s.zkw.WatchChildren("/jfs-root/extnode-rg/" + strconv.Itoa(s.rg))
		if e != nil {
			//close the handle in case of errors
			s.zkw.Close()
			s.zkw = nil
			s.errLog.Println("Fatal: failed to watch Zk")
			<-time.After(AbnormalWakeUpPeriod * time.Minute)
			continue
		}

		select {
		case <-ch:
			s.infoLog.Println("Info: notified by Zk")
			continue
		case <-time.After(NormalWakeUpPeriod * time.Minute):
			s.infoLog.Println("Info: timeout to check group changes")
			continue
		}
	}
}
